#include "app_pool.h"
#include "log/log.h"
#include <mqueue.h>
#include <pthread.h>
#include <stdlib.h>
static char *mq_name = "/mq_test";
static mqd_t mq_fd;

//初始化线程池大小
static int thread_num;
static pthread_t *thread_pool;//线程池的指针
/**
 * 线程函数
 *  不断的从任务队列中获取任务并执行
 */
void *thread_fun(void *arg){
 Task task;
    while (1)
    {
        ssize_t len = mq_receive(mq_fd, (char *)&task, sizeof(Task), NULL);
        if (len == sizeof(Task))
        {
            task.task_fun(task.arg);
        }
    }
}

//初始化线程池/任务队列
int app_pool_init(int size){
    struct mq_attr attr;
    attr.mq_maxmsg = 10;//队列中最大消息数
    attr.mq_msgsize = sizeof(Task);

    mq_fd = mq_open(mq_name, O_CREAT | O_RDWR, 0644,&attr);
    log_debug("mq_fd = %d", mq_fd);
    if(mq_fd == -1){
        log_error("mq_open error");
        return -1;
    }
    thread_num = size;
    thread_pool = (pthread_t *)malloc(sizeof(pthread_t)*thread_num);

    for (int i = 0; i < thread_num; i++)
    {
        pthread_create(&thread_pool[i],NULL,thread_fun,NULL);
    }
    return 0;
}

//销毁线程池/任务队列
void app_pool_close(){

    for (int i = 0; i < thread_num; i++)
    {
        pthread_cancel(thread_pool[i]);
        pthread_join(thread_pool[i],NULL);
    }

    //释放线程池内存
    free(thread_pool);

    //删除消息队列
    mq_unlink(mq_name);
}

//向线程池中添加任务

int app_pool_registerTask(int (*task_fun)(void *),void *arg){
    Task task = {
        .task_fun = task_fun,
        .arg = arg
    };

    mq_send(mq_fd,(char *)&task,sizeof(Task),0);

    return 0;
}